package com.atguigu.Flink.function;

import com.atguigu.Flink.POJO.Event;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import javax.lang.model.element.VariableElement;
import java.sql.Time;
import java.util.Random;
import java.util.concurrent.TimeUnit;

//自定义数据源
public class ClickSource implements SourceFunction<Event> {
    private boolean isFlag = true;
    String[] users = {"otto","xuanDog","shinyRuo","BZZB","shenhaiFish"};
    String[] urls = {"/douyu","/huya","/bilibili","PornHub"};
    Random random = new Random();

    @Override
    public void run(SourceContext<Event> sourceContext) throws Exception {
        while(isFlag){
            String user = users[random.nextInt(users.length)];
            String url = urls[random.nextInt(urls.length)];
            Long ts = System.currentTimeMillis();
            Event event = new Event(user,url,ts);

            //发送数据到flink
            sourceContext.collect(event);
            //控制一秒生成一条
            //Thread.sleep(1000);
            TimeUnit.SECONDS.sleep(1);
        }
    }

    @Override
    public void cancel() {
        isFlag=false;
    }
}
